package com.foreveross.fetl.core.domain.freight.repository.impl;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.mortbay.log.Log;
import org.springframework.stereotype.Repository;

import com.foreveross.fetl.common.HBaseTable;
import com.foreveross.fetl.common.hbase.client.HBaseHelper;
import com.foreveross.fetl.common.jpa.EntityHibernateRepository;
import com.foreveross.fetl.core.domain.freight.AirSpread;
import com.foreveross.fetl.core.domain.freight.repository.IAirSpreadRepository;
import com.foreveross.fetl.util.DateUtils;
import com.foreveross.fetl.util.RowkeyUtils;
import com.foreveross.fetl.util.StringUtils;

@Repository("airSpreadRepository")
public class AirSpreadRepositoryImpl extends EntityHibernateRepository
		implements IAirSpreadRepository {

	@Override
	public List<AirSpread> findByCondition(Date reportDate, Date flightDate,
			String fromCity, String toCity, String flightNo, String hbaseKey,
			String channelKey) {
		StringBuffer jpql =new StringBuffer("select airSpread from AirSpread airSpread where 1=1 ");
		List<Object> conditionVals = new ArrayList<Object>();
		if(reportDate!=null){
			jpql.append(" and airSpread.reportDate =? ");
			conditionVals.add(reportDate);
		}
		if(flightDate !=null){
			jpql.append(" and airSpread.flightDate =? ");
			conditionVals.add(flightDate);
		}
		if(null!=fromCity && !"".equals(fromCity)){
			jpql.append(" and airSpread.fromCity =? ");
			conditionVals.add(fromCity);
		}
		if(null!=toCity && !"".equals(toCity)){
			jpql.append(" and airSpread.toCity =? ");
			conditionVals.add(toCity);
		}
		if(null!=flightNo && !"".equals(flightNo)){
			jpql.append(" and airSpread.flightNo =? ");
			conditionVals.add(flightNo);
		}
		if(null!=hbaseKey && !"".equals(hbaseKey)){
			jpql.append(" and airSpread.hbaseKey =? ");
			conditionVals.add(hbaseKey);
		}
		if(null!=channelKey && !"".equals(channelKey)){
			jpql.append(" and airSpread.channelKey =? ");
			conditionVals.add(channelKey);
		}
		jpql.append(" order by airSpread.creattime desc ");
		return super.queryResult(jpql.toString(), conditionVals.toArray());
	}

	@Override
	public void saveOrUpdate(AirSpread airSpread) {
		super.save(airSpread);

	}

	@Override
	public void findByID(Long id) {
		super.get(AirSpread.class, id);

	}

	@Override
	public int createSingle(List<String> airCompanys, List<String> compareChannels, String fromCity,
			String toCity, Date reportDate, Date flightDate) {
		List<AirSpread> listSpread =new ArrayList<AirSpread>();
//		PortEntity portEntity =new PortEntity();
//		portEntity=portEntity.findByFromCityCode(fromCity);
//		if(portEntity==null){
//			return 0;
//		}
		//航空公司的价格集合
		Map<String, List<AirSpread>> companySpreadMap = new HashMap<String, List<AirSpread>>();
		for(String channel : airCompanys){
			//这里不做过多的业务逻辑判断，组装好数据即返回，快速释放hbase连接对象
			companySpreadMap.put(channel, this.buildChannelAireSpread(channel,  fromCity, toCity, reportDate, flightDate));
			//添加今天抓取的 
			companySpreadMap=this.todayGrabList(channel, companySpreadMap, reportDate,flightDate,fromCity, toCity);
		}
		//渠道的价格集合
		Map<String, List<AirSpread>> channelSpreadMap = new HashMap<String, List<AirSpread>>();
		for(String channel:compareChannels){
			//这里不做过多的业务逻辑判断，组装好数据即返回，快速释放hbase连接对象
			channelSpreadMap.put(channel, this.buildChannelAireSpread(channel,  fromCity, toCity, reportDate, flightDate));
		}
		//以航空公司的航班为基准的价差对比
		//因为模型对象是以一个airSpread对象存储的，所以循环渠道价格集合,交叉匹配航空公司价格集合
		AirSpread tempAs = null;
		List<AirSpread> tempAirSpreads = null;		
		for(String channel : airCompanys){
			tempAirSpreads = companySpreadMap.get(channel);
			if(tempAirSpreads == null) continue;
			for(AirSpread as : tempAirSpreads){ //航空公司下有这些航班
				boolean hasChannel = false;
				for(String c : compareChannels){
					tempAs = this.getCompareChannelAirSpread(channelSpreadMap.get(c), 
							as.getFlightNo());
					if(tempAs != null){
						//这里将值调过来，设置成了对应航空公司下航班价格
						tempAs.setChannelKey(c);
						tempAs.setChannelPrice(tempAs.getPrice());
						tempAs.setHbaseKey(channel);
						tempAs.setPrice(as.getPrice());
						
						listSpread.add(tempAs);
						hasChannel = true;
					}
				}
				//航班下没有一个渠道对应则创建一个自身
				if(!hasChannel) listSpread.add(as);
			}
			/*	List<AirSpread> channelAir=channelSpreadMap.get(channel);
				for(AirSpread airSp: channelAir){
				if(!this.isChannelAirSpreadExit(listSpread, airSp.getFlightNo(), airSp.getChannelKey())){
					airSp.setChannelPrice(airSp.getPrice());
					airSp.setPrice(null);
					listSpread.add(airSp);
				}
			}*/
		}
		
		return this.saveList(listSpread);
	}
	
	private AirSpread getCompareChannelAirSpread(List<AirSpread> results, String flightNo){
		if(results != null){
			for(AirSpread as : results){
				if(flightNo.equalsIgnoreCase(as.getFlightNo())){
					return as;
				}
			}
		}
		return null;
	}
	private boolean isAirSpreadExit(List<AirSpread> results, String flightNo){
		if(results != null){
			for(AirSpread as: results){
				if(flightNo.equalsIgnoreCase(as.getFlightNo())){
					return true;
				}
			}
		}
		return false;
	}
	
	private List<AirSpread> buildChannelAireSpread(String channel, String fromCity,
			String toCity, Date reportDate, Date flightDate) {
		List<AirSpread> listSpread =new ArrayList<AirSpread>();
		String startRowKey =RowkeyUtils.getRowKey(reportDate, channel,  fromCity, toCity, flightDate, RowkeyUtils.START_ROW_KEY);
		String endRowKey=RowkeyUtils.getRowKey(reportDate, channel,  fromCity, toCity, flightDate, RowkeyUtils.STOP_ROW_KEY);
		ResultScanner rs = null;
		try {
			rs = HBaseHelper.getRecords(HBaseTable.BASE_DATA_FLIGHT,
					startRowKey, endRowKey);
			for (Result res : rs) {
				listSpread.add(this.buildAirSpread(res,channel,reportDate, flightDate));
			}
		} catch (Exception e) {
			Log.info(StringUtils.getStackTrace(e));
		}finally{
			if(rs != null) rs.close();
			rs = null;
		}
		return listSpread;
	}

	private AirSpread buildAirSpread(Result res, String chanel, Date reportDate, Date flightDate){
		Map<String, Map<String, String>> curMap = HBaseHelper
				.transformHBaseRowToMap(res);
		Map<String, String> curPalneInfoMap = curMap.get("planeinfo");
		double flightPrice = Float.parseFloat(curPalneInfoMap.get("lowerPrice") != null 
				 ? curPalneInfoMap.get("lowerPrice") : "0");
		if (flightPrice <= 0) {
			return null;
		}
		if(curPalneInfoMap.get("flightId")==null) return null;
		if(curPalneInfoMap.get("flightId").indexOf("|")>0){
			curPalneInfoMap.put("flightId", curPalneInfoMap.get("flightId").substring(0, curPalneInfoMap.get("flightId").indexOf("|")));
		}
		if(curPalneInfoMap.get("flightId").indexOf(",")>0){
			curPalneInfoMap.put("flightId", curPalneInfoMap.get("flightId").substring(0, curPalneInfoMap.get("flightId").indexOf(",")));
		}
		String flightNo =curPalneInfoMap.get("flightId");
		String flightTimeType=curPalneInfoMap.get("flightBatchStartTime").substring(0, 12);
		Date startTime =DateUtils.parse(flightTimeType, "yyyyMMddHHmm");
		AirSpread airSpread =new AirSpread();
			airSpread.setReportDate(DateUtils.parse(DateUtils.format(reportDate, "yyyyMMdd"), "yyyyMMdd"));
			airSpread.setFlightDate(DateUtils.parse(DateUtils.format(flightDate, "yyyyMMdd"), "yyyyMMdd"));
			airSpread.setFromCity(curPalneInfoMap.get("fromCity"));
			airSpread.setToCity(curPalneInfoMap.get("toCity"));
			airSpread.setFromCityName(curPalneInfoMap.get("fromCityName"));
			airSpread.setToCityName(curPalneInfoMap.get("toCityName"));
			airSpread.setStartTime(startTime);
			airSpread.setCreattime(new Date());
			airSpread.setFlightNo(flightNo);
			airSpread.setChannelKey(chanel);
			airSpread.setPrice(flightPrice);
		return airSpread;
	}
	 /** 
     * 对double数据进行取精度. 
     * @param value  double数据. 
     * @param scale  精度位数(保留的小数位数). 
     * @param roundingMode  精度取值方式. 
     * @return 精度计算后的数据. 
     */ 
    public static double round(double value, int scale,
             int roundingMode) {  
        BigDecimal bd = new BigDecimal(value);  
        bd = bd.setScale(scale, roundingMode);  
        double d = bd.doubleValue();  
        bd = null;  
        return d;  
    }
    
	
	
	public int saveList(List<AirSpread> list){
		if(list==null || list.size()<=0){
			   return 0;
		   }
		   int i=0;
		  for(AirSpread en :list){
			List<AirSpread> oldList = this.findByCondition(en.getReportDate(),
					en.getFlightDate(), en.getFromCity(), en.getToCity(),
					en.getFlightNo(), en.getHbaseKey(), en.getChannelKey());
			if(oldList!=null && !oldList.isEmpty()){
				AirSpread oldEntity =oldList.get(0);
				Calendar c = Calendar.getInstance();
				c.add(Calendar.HOUR_OF_DAY, -2);
				Calendar lastGreatTime = Calendar.getInstance();
				lastGreatTime.setTime(oldEntity.getCreattime());
				if(lastGreatTime.after(c)){
					if(en.getPrice()==null){
						en.setPrice(oldEntity.getPrice());
					}
					if(en.getChannelPrice()==null){
						en.setChannelKey(oldEntity.getChannelKey());
						en.setChannelPrice(oldEntity.getChannelPrice());
					}
				}
				en.setId(oldEntity.getId());
			}
			if(en.getPrice()!=null && en.getPrice()!=0 && en.getChannelPrice()!=null && en.getChannelPrice()!=0){
				en.setSpreadAmount(en.getPrice()-en.getChannelPrice());
				en.setSpread(round(en.getSpreadAmount()/en.getPrice(),4,BigDecimal.ROUND_HALF_UP));
			}
			this.save(en);
			   i++;
		  }
		return i;
	}
	
	 private Map<String, List<AirSpread>> todayGrabList(String hbaseKey,Map<String, List<AirSpread>> grabMap,Date reportDate,Date flightDate,String fromCity,
				String toCity){
		 List<AirSpread> list =new ArrayList<AirSpread>();
		 list =grabMap.get(hbaseKey);
		 StringBuffer jpql =new StringBuffer("select airSpread from AirSpread airSpread where 1=1 ");
			List<Object> conditionVals = new ArrayList<Object>();
			
			jpql.append(" and airSpread.reportDate =? ");
			conditionVals.add(DateUtils.parse(DateUtils.format(reportDate, "yyyyMMdd"), "yyyyMMdd"));
			
			if(flightDate !=null){
				jpql.append(" and airSpread.flightDate =? ");
				conditionVals.add(DateUtils.parse(DateUtils.format(flightDate, "yyyyMMdd"), "yyyyMMdd"));
			}
			
			if(null!=hbaseKey && !"".equals(hbaseKey)){
				jpql.append(" and airSpread.hbaseKey =? ");
				conditionVals.add(hbaseKey);
			}
			if(null!=fromCity && !"".equals(fromCity)){
				jpql.append(" and airSpread.fromCity =? ");
				conditionVals.add(fromCity);
			}
			if(null!=toCity && !"".equals(toCity)){
				jpql.append(" and airSpread.toCity =? ");
				conditionVals.add(toCity);
			}
			List<AirSpread> listSpread=super.queryResult(jpql.toString(), conditionVals.toArray());
			for(AirSpread airSpread : listSpread){
				if(!this.isAirSpreadExit(grabMap.get(hbaseKey), airSpread.getFlightNo())){
					list.add(airSpread);
				}
			}
			grabMap.put(hbaseKey, list);
		 return grabMap;
	}
}
